package com.bs.flinktest.controller;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bs.flinktest.vo.TestDataSource;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.core.execution.JobClient;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutionException;

/**
 * @author bingshao
 * @date 2023/2/1
 **/
@RestController
@RequestMapping("/test")
public class TestController {

//    @Resource
//    private ClusterClient restClusterClient;

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @GetMapping("/test")
    public void test() throws ExecutionException, InterruptedException {
        String s = stringRedisTemplate.opsForValue().get("1");
        JobClient jobClient = JSON.parseObject(s, JobClient.class);
        System.out.println("关闭前" + jobClient.getJobStatus().get());
        jobClient.cancel();
        System.out.println("关闭后" + jobClient.getJobStatus().get());
    }


    @GetMapping("/test/{id}")
    public void test(@PathVariable("id") String id) throws ExecutionException, InterruptedException {
        JobID jobID = new JobID(id.getBytes(StandardCharsets.UTF_8));
//        CompletableFuture future = restClusterClient.cancel(jobID);
//        System.out.println("取消======"+future.get());

    }

    @GetMapping("/test1")
    public String test1() throws ExecutionException, InterruptedException {
        JobID jobID = TestDataSource.jobClientList.get(0).getJobID();
        System.out.println("jobID===========>" + jobID);
        TestDataSource.jobClientList.get(0).cancel();
        System.out.println("取消任务===========>");

        JobStatus jobStatus = TestDataSource.jobClientList.get(0).getJobStatus().get();
        System.out.println("取消任务之后的状态===========>" + jobStatus);
        return "ok";
    }

    public static void main(String[] args) {
        String s1 = "aaaaaaaa";
        Object parse = JSONObject.parse("sawsaasxdx");
        System.out.println();
    }

}